package com.pw.study.flink.state;

import org.apache.flink.api.common.state.BroadcastState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.state.ReadOnlyBroadcastState;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;

public class OperatorBroadcastState {

    public static void main(String[] args) {
        stateList();
    }

    private static void stateList() {
        Configuration conf = new Configuration();
        conf.setInteger("rest.port", 20000);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
        env.setParallelism(2);
        /*
        广播状态使用的套路:
        1. 需要两个流: 1个是数据流  2 配置流. -配置流中的数据用来控制数据流中的数据的处理方式
        2. 把配置流做成广播流
        3. 让数据流去connect广播流
        4. 把广播流中的数据放入到广播状态
        5. 数据流中的数据处理的时候, 从广播状态读取配置信息
         */
        //输入数据,1.数据流，2.配置流
        DataStreamSource<String> dataStream = env.socketTextStream("hadoop111", 8888);
        DataStreamSource<String> configStream = env.socketTextStream("hadoop111", 9999);
        MapStateDescriptor<String, String> configDesc = new MapStateDescriptor<>("configState", String.class, String.class);
        BroadcastStream<String> stream = configStream.broadcast(configDesc);

        dataStream.connect(stream).process(new BroadcastProcessFunction<String, String, String>() {
            @Override
            public void processElement(String value, ReadOnlyContext ctx, Collector<String> out) throws Exception {
                ReadOnlyBroadcastState<String, String> bcState = ctx.getBroadcastState(configDesc);
                String config = bcState.get("switch");
                if ("1".equals(config)) {
                    out.collect("数据：A " + value + " 使用1号逻辑处理。。。");
                } else if ("2".equals(config)) {
                    out.collect("数据：B " + value + " 使用2号逻辑处理。。。");
                } else if ("3".equals(config)) {
                    out.collect("数据C:  " + value + " 使用3号逻辑处理。。。");
                } else {
                    out.collect("数据other： " + value + " 使用其他数据逻辑处理。。。");
                }
            }

            @Override
            public void processBroadcastElement(String value, Context ctx, Collector<String> out) throws Exception {
                BroadcastState<String, String> bcState = ctx.getBroadcastState(configDesc);
                bcState.put("switch", value);
            }
        }).print();


        try {
            env.execute();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
